package org.leiyang.common.holder;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.leiyang.common.dtos.MsgServerInfo;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

import static org.leiyang.common.util.Constants.online_server_key;

/**
 * 管理在线的server列表
 */
@Component
public class OnlineServerHolder {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final ReactiveRedisCache redisCache;
    private MsgServerAndClientHolder msgServerAndClientHolder;
    /**
     * 存储当前Msg_Server下运行的所有channel
     * 格式: channelId:channel
     */
    private static final Map<String, Channel> ONLINE_SERVER_CHANNELS_MAP = new ConcurrentHashMap<>();

    public OnlineServerHolder(ReactiveRedisCache redisCache) {
        this.redisCache = redisCache;
    }

    public MsgServerAndClientHolder getMsgServerAndClientHolder() {
        return msgServerAndClientHolder;
    }
    @Autowired
    public void setMsgServerAndClientHolder(MsgServerAndClientHolder msgServerAndClientHolder) {
        this.msgServerAndClientHolder = msgServerAndClientHolder;
    }

    /**
     * 上线的server加入
     * 使用场景: server启动成功时调用 或者 启动之后获取对应的MsgServer信息
     * @param address ip
     * @param port 端口
     */
    public Mono<MsgServerInfo> joinOnlineServer(String address, String port) {
        final String redisKey = online_server_key.concat(address.replaceAll("\\.", "_")).concat("&").concat(port);
        MsgServerInfo newMsgServerInfo = MsgServerInfo.builder().serverAddress(address)
                .serverPort(port)
                .channelIds(new ArrayList<>())
                .build();
        return redisCache.setCacheObjectNX(redisKey, newMsgServerInfo, -1).flatMap(setResult -> {
            if(setResult) {
                return Mono.just(newMsgServerInfo);
            }
            return redisCache.getCacheObject(redisKey).map(obj -> (MsgServerInfo) obj);
        });
    }

    /**
     * 获取对应的channel
     * @param channelId channelID
     */
    public Channel gotCachedChannel(String channelId) {
        return ONLINE_SERVER_CHANNELS_MAP.get(channelId);
    }

    /**
     * 缓存的ChannelId列表
     */
    public Set<String> cachedChannelIds() {
        return ONLINE_SERVER_CHANNELS_MAP.keySet();
    }

    /**
     * 剔除离线的server
     * 使用场景: server下线时调用
     * @param address ip
     * @param port 端口
     */
    public Mono<Boolean> removeOffLineServer(String address, String port) {
        final String redisKey = online_server_key.concat(address.replaceAll("\\.", "_")).concat("&").concat(port);
        Mono<MsgServerInfo> serverInfoMonoGet = redisCache.getCacheObject(redisKey).map(obj -> (MsgServerInfo)obj);
        Mono<Long> serverInfoMonoDel = redisCache.deleteObject(redisKey);
        return serverInfoMonoGet.zipWhen(msgServerInfo -> serverInfoMonoDel, (info, delNum) -> {
            if(Objects.equals(delNum, 1L)) {
                msgServerAndClientHolder.batchRemoveByServer(info);
            }
            return true;
        });
    }

    /**
     * 加入channel
     * 使用场景: 客户端发起连接成功时调用
     * @param address ip
     * @param port 端口
     * @param ctx ctx
     */
    public Mono<Boolean> joinChannel(String address, String port, ChannelHandlerContext ctx) {
        final String redisKey = online_server_key.concat(address.replaceAll("\\.", "_")).concat("&").concat(port);
        return redisCache.getCacheObject(redisKey).flatMap(obj -> {
            if(Objects.nonNull(obj)) {
                String channelId = ctx.channel().id().asLongText();
                ONLINE_SERVER_CHANNELS_MAP.put(channelId, ctx.channel());
                MsgServerInfo msgServerInfo = (MsgServerInfo) obj;
                msgServerInfo.getChannelIds().add(channelId);
                return redisCache.setCacheObject(redisKey, msgServerInfo);
            }
            return Mono.just(true);
        });
    }

    /**
     * 移除channel
     * 使用场景: 客户端与服务端主动断连时
     * @param address ip
     * @param port 端口
     * @param channel ctx
     */
    public Mono<Boolean> removeChannel(String address, String port, Channel channel) {
        final String redisKey = online_server_key.concat(address.replaceAll("\\.", "_")).concat("&").concat(port);
        return redisCache.getCacheObject(redisKey).flatMap(obj -> {
            if(Objects.nonNull(obj)) {
                String channelId = channel.id().asLongText();
                ONLINE_SERVER_CHANNELS_MAP.remove(channelId);
                MsgServerInfo msgServerInfo = (MsgServerInfo) obj;
                msgServerInfo.getChannelIds().remove(channelId);
                return redisCache.setCacheObject(redisKey, msgServerInfo);
            }
            return Mono.just(true);
        });
    }

    /**
     * 获取当前连接数最少的server
     * @return 当前连接数最少的server
     */
    public Mono<MsgServerInfo> chooseMostIdleServer() {
        return redisCache.keys(online_server_key.concat("*"))
                .flatMap(redisCache::getCacheObject)
                .filter(Objects::nonNull)
                .map(obj -> (MsgServerInfo)obj)
                .collectList()
                .map(msgServerInfos -> {
                    AtomicReference<MsgServerInfo> result = new AtomicReference<>();
                    msgServerInfos.stream()
                            .min(Comparator.comparing(msgServerInfo -> msgServerInfo.getChannelIds().size()))
                            .ifPresent(result::set);
                    MsgServerInfo target = result.get();
                    logger.info("本次chooseMostIdleServer获取到的MsgServer信息: {}", target.getServerAddress().concat(":").concat(target.getServerPort()));
                    //客户端不需要该字段信息
                    if(Objects.nonNull(target)) {
                        target.setChannelIds(null);
                    }
                    return target;
                });
    }

}